package com.fwmagic.flinkcdc.oracle;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @description:
 * @author: fangwei
 * @date: 2022/5/19
 **/
public class FlinkSQLCDC {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建 Flink-Oracle-CDC
        tableEnv.executeSql("CREATE TABLE f_user (\n" +
                "    db_name STRING METADATA FROM 'database_name' VIRTUAL,\n" +
                "    schema_name STRING METADATA FROM 'schema_name' VIRTUAL, \n" +
                "    table_name STRING METADATA  FROM 'table_name' VIRTUAL,\n" +
                "    id INT NOT NULL,\n" +
                "    name STRING,\n" +
                "    sex STRING,\n"+
                "    PRIMARY KEY(id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'oracle-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '1521',\n" +
                "    'username' = 'family',\n" +
                "    'password' = 'chickenkang',\n" +
                "    'database-name' = 'XE',\n" +
                "    'schema-name' = 'family',\n" +
                "    'table-name' = 'family.t_user'\n" +
                ")");

        Table table = tableEnv.sqlQuery("select * from f_user");

        DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);

        retractStream.print();


        env.execute("FlinkSQL-CDC oracle binlog");
    }
}
